package com.hugoo.core.db.pools;

import java.sql.Connection;
import java.sql.SQLException;
import java.util.LinkedList;
import java.util.Queue;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import javax.sql.ConnectionEvent;
import javax.sql.ConnectionEventListener;
import javax.sql.ConnectionPoolDataSource;
import javax.sql.PooledConnection;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

public class MCPM {

    private ConnectionPoolDataSource dataSource;
    private int maxConnections;
    private int timeout;
    private Semaphore semaphore;
    private Queue<PooledConnection> recycledConnections;
    private int activeConnections;
    private PoolConnectionEventListener poolConnectionEventListener;
    private boolean isDisposed;

    /**
     */
    public static class TimeoutException extends RuntimeException {

        private static final long serialVersionUID = 1;

        public TimeoutException() {
            super("Timeout while waiting for a free database connection.");
        }
    }

    public MCPM(ConnectionPoolDataSource dataSource, int maxConnections) {
        this(dataSource, maxConnections, 60);//设置超时时间为60秒
    }

    public MCPM(ConnectionPoolDataSource dataSource, int maxConnections, int timeout) {
        this.dataSource = dataSource;
        this.maxConnections = maxConnections;
        this.timeout = timeout;
        if (maxConnections < 1) {
            throw new IllegalArgumentException("Invalid maxConnections value.");
        }
        semaphore = new Semaphore(maxConnections, true);
        recycledConnections = new LinkedList<PooledConnection>();
        poolConnectionEventListener = new PoolConnectionEventListener();
    }

    public synchronized void dispose() throws SQLException {
        log.info("call dispose.");
        if (isDisposed) {
            return;
        }
        log.info("start dispose.");
        isDisposed = true;
        SQLException e = null;
        while (!recycledConnections.isEmpty()) {
            PooledConnection pconn = recycledConnections.remove();
            try {
                pconn.close();
            } catch (SQLException e2) {
                if (e == null) {
                    e = e2;
                }
            }
        }
        if (e != null) {
            throw e;
        }
    }

    public Connection getConnection() throws SQLException {
        synchronized (this) {
            if (isDisposed) {
                throw new IllegalStateException("Connection pool has been disposed.");
            }
        }
        try {
            if (!semaphore.tryAcquire(timeout, TimeUnit.SECONDS)) {
                throw new TimeoutException();
            }
        } catch (InterruptedException e) {
            throw new RuntimeException("Interrupted while waiting for a database connection.", e);
        }
        boolean ok = false;
        try {
            Connection conn = getConnection2();
            ok = true;
            return conn;
        } finally {
            if (!ok) {
                semaphore.release();
            }
        }
    }
    protected final static Log log = LogFactory.getLog(MCPM.class);

    private synchronized Connection getConnection2() throws SQLException {
        if (isDisposed) {
            throw new IllegalStateException("Connection pool has been disposed.");   // test again with lock
        }
        PooledConnection pconn;
        if (!recycledConnections.isEmpty()) {
            pconn = recycledConnections.remove();
        } else {
            pconn = dataSource.getPooledConnection();
        }
        Connection conn = pconn.getConnection();
        activeConnections++;
        pconn.addConnectionEventListener(poolConnectionEventListener);
        assertInnerState();
        return conn;
    }

    private synchronized void recycleConnection(PooledConnection pconn) {
        if (isDisposed) {
            disposeConnection(pconn);
            return;
        }
        if (activeConnections <= 0) {
            throw new AssertionError();
        }
        activeConnections--;
        semaphore.release();
        recycledConnections.add(pconn);
        assertInnerState();
    }

    private synchronized void disposeConnection(PooledConnection pconn) {
        if (activeConnections <= 0) {
            throw new AssertionError();
        }
        activeConnections--;
        semaphore.release();
        closeConnectionNoEx(pconn);
        assertInnerState();
    }

    private void closeConnectionNoEx(PooledConnection pconn) {
        try {
            pconn.close();
        } catch (Exception e) {
            log.warn("Error while closing database connection: ", e);
        }
    }

    private void assertInnerState() {
        if (activeConnections < 0) {
            throw new AssertionError();
        }
        if (activeConnections + recycledConnections.size() > maxConnections) {
            throw new AssertionError();
        }
        if (activeConnections + semaphore.availablePermits() > maxConnections) {
            throw new AssertionError();
        }
    }

    private class PoolConnectionEventListener implements ConnectionEventListener {

        @Override
        public void connectionClosed(ConnectionEvent event) {
            PooledConnection pconn = (PooledConnection) event.getSource();
            pconn.removeConnectionEventListener(this);
            recycleConnection(pconn);
        }

        @Override
        public void connectionErrorOccurred(ConnectionEvent event) {
            PooledConnection pconn = (PooledConnection) event.getSource();
            pconn.removeConnectionEventListener(this);
            disposeConnection(pconn);
        }
    }

    public synchronized int getActiveConnections() {
        return activeConnections;
    }
}
